package com.wisdytech.linkdcs.configuration.service.impl;

import com.alibaba.fastjson.JSON;
import com.wisdytech.common.bean.EquipParamDataDTO;
import com.wisdytech.common.bean.ParamDataDto;
import com.wisdytech.common.enums.ConfigurationWebSocketType;
import com.wisdytech.linkdcs.configuration.dao.ITtEqConfigurationDao;
import com.wisdytech.linkdcs.configuration.dto.ConfigurationEqDto;
import com.wisdytech.linkdcs.configuration.dto.ConfigurationParamDto;
import com.wisdytech.linkdcs.configuration.dto.WebSocketDto;
import com.wisdytech.linkdcs.configuration.model.TtEqConfiguration;
import com.wisdytech.linkdcs.configuration.service.ITtEqConfigurationDataService;
import com.wisdytech.linkdcs.equipment.dto.EquipAlarmDataDto;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Service
@Transactional
public class TtEqConfigurationDataServiceImpl implements ITtEqConfigurationDataService {


    private static final Logger logger = LoggerFactory.getLogger(TtEqConfigurationDataServiceImpl.class);

    private static final Lock pushEquipDataLock = new ReentrantLock();

    /**
     * restful服务
     */
    /*@Autowired
    private RestTemplate restTemplate;*/

    @Autowired
    private ITtEqConfigurationDao ttEqConfigurationDao;

    @Autowired
    private HbaseTemplate hbaseTemplate;

    //设备数据表
    @Value("${hbase.tableName}")
    private String TABLE_NAME = "MachineData";

    @Value("${push.test}")
    private Boolean pushTest;

    @Value("${hbase.time-zone}")
    private Boolean timeZone;


    /**
     * @param id 组态记录id
     * @return
     */
    @Override
    public List<ConfigurationEqDto> getConfigurationEqLists(String id) {
        TtEqConfiguration ttEqConfiguration = ttEqConfigurationDao.selectById(id);
        if (null == ttEqConfiguration) {
            return Collections.emptyList();
        }
        String jsonFile = ttEqConfiguration.getJsonEquips();
        if (StringUtils.isBlank(jsonFile)) {
            return Collections.emptyList();
        }
        //解析设备json
        try {
            return JSON.parseArray(jsonFile, ConfigurationEqDto.class);
        } catch (Exception e) {
            logger.error("解析组态设计json失败！！组态记录为" + ttEqConfiguration.getPkId() + "数据格式为" + jsonFile, e);
            return Collections.emptyList();
        }
    }


    @Override
    public void pushAlarmEquipmentData(List<EquipAlarmDataDto> alarmMsg, WebSocketSession session) {
        WebSocketDto dto = new WebSocketDto(
                ConfigurationWebSocketType.ALARM_DATA.getCode(), alarmMsg);
        TextMessage meg = new TextMessage(
                JSON.toJSONString(dto));
        try {
            //推送数据至终端
            if (session.isOpen()) {
                session.sendMessage(meg);
            }
        } catch (Exception e) {
            logger.error("推送websocket失败", e);
        }

    }

    @Override
    public void pushEquipmentData(List<ConfigurationEqDto> eqDtos, Channel channel) {
        WebSocketDto dto = new WebSocketDto(ConfigurationWebSocketType.EQUIP_DATA.getCode(), eqDtos);
        TextWebSocketFrame msg = new TextWebSocketFrame(JSON.toJSONString(dto));
        if (channel.isActive()) {
            channel.writeAndFlush(msg);
        }
    }

    /**
     * 推送设备实时数据
     *
     * @param eqDtos
     * @param session
     */
    @Override
    public void pushEquipmentData(List<ConfigurationEqDto> eqDtos, WebSocketSession session) {
        WebSocketDto dto = new WebSocketDto(ConfigurationWebSocketType.EQUIP_DATA.getCode(), eqDtos);
        TextMessage msg = new TextMessage(JSON.toJSONString(dto));
        pushEquipDataLock.lock();
        try {
            //推送数据至终端
            if (session.isOpen()) {
                session.sendMessage(msg);
            }
        } catch (Exception e) {
            logger.error("推送websocket失败", e);
        } finally {
            pushEquipDataLock.unlock();
        }


    }

    @Override
    public List<ConfigurationEqDto> getEquipmentRealData(String configurationId) {
        //获取组态信息
        List<ConfigurationEqDto> eqDtos = getConfigurationEqLists(configurationId);
        List<ConfigurationEqDto> eqDtoDatas = new ArrayList<>();
        if (eqDtos == null || eqDtos.size() == 0) {
            return eqDtoDatas;
        }
        // 计算当前时间的时间戳（+8小时）
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(new Date());
        if (timeZone) {
            calendar.add(Calendar.HOUR_OF_DAY, 8);
        }
        long startTime = calendar.getTime().getTime();
        // 往前取5分钟
        long endTime = startTime - 1000L * 60L * 5L;
        eqDtos.forEach(eqDto -> {
            if (pushTest) {
                eqDtoDatas.add(getEquipDataByEquipNoAndParamTest(eqDto, startTime, endTime));
            } else {
                eqDtoDatas.add(getEquipDataByEquipNoAndParam(eqDto, startTime, endTime));
            }
        });
        logger.info("equip and param info:{}", JSON.toJSONString(eqDtoDatas));
        return eqDtoDatas;
    }

    /**
     * 根据设备编码和参数查询数据
     *
     * @return
     */
    private ConfigurationEqDto getEquipDataByEquipNoAndParamTest(ConfigurationEqDto eqDto, long startTime, long endTime) {
        ConfigurationEqDto data = new ConfigurationEqDto();
        if (eqDto == null) {
            logger.warn(" json is null");
            return data;
        }
        if (StringUtils.isBlank(eqDto.getEquipNo())) {
            logger.warn(" equipNo is null");
            return data;
        }
        if (eqDto.getProperties() == null || eqDto.getProperties().size() == 0) {
            logger.warn(" equip param is null");
            return data;
        }
        logger.info("查询设备实时数据,查询参数为：equipNo -->{}, params -->{} ", eqDto.getEquipNo(), JSON.toJSONString(eqDto.getProperties()));

        // param.getParamId 即为列名
        for (ConfigurationParamDto param : eqDto.getProperties()) {//查询扫描器
            param.setValue(getValue());
        }
        //TODO 目前还未确认上线时间
        eqDto.setStatus("0");
        return eqDto;
    }

    private double getValue() {
        Random random = new Random();
        int value = random.nextInt(100);
        BigDecimal bg = new BigDecimal(value).setScale(2, RoundingMode.HALF_UP);
        return bg.doubleValue();
    }

    /**
     * 根据设备编码和参数查询数据
     *
     * @return
     */
    private ConfigurationEqDto getEquipDataByEquipNoAndParam(ConfigurationEqDto eqDto, long startTime, long endTime) {
        ConfigurationEqDto data = new ConfigurationEqDto();
        if (eqDto == null) {
            logger.warn(" json is null");
            return data;
        }
        if (StringUtils.isBlank(eqDto.getEquipNo())) {
            logger.warn(" equipNo is null");
            return data;
        }
        if (eqDto.getProperties() == null || eqDto.getProperties().size() == 0) {
            logger.warn(" equip param is null");
            return data;
        }
        logger.info("查询设备实时数据,查询参数为：equipNo -->{}, params -->{} ", eqDto.getEquipNo(), JSON.toJSONString(eqDto.getProperties()));

        // param.getParamId 即为列名
        for (ConfigurationParamDto param : eqDto.getProperties()) {//查询扫描器
            // 列名
            String columnName = getColumnName(eqDto.getEquipNo(), param.getParamId());
            if (StringUtils.isBlank(columnName)) {
                continue;
            }
            Scan scan = new Scan();
            //添加列簇
            //根据参数名和equipNo获取列簇名
            byte[] columnFamily = getColumnFamily(eqDto.getEquipNo(), param.getParamId());
            if (columnFamily == null) {
                continue;
            }
            scan.addFamily(columnFamily);

            //分页查询条件
            FilterList filterList = new FilterList();
            //分页过滤
            Filter filter = new PageFilter(1);
            filterList.addFilter(filter);
            scan.setFilter(filter);
            //开始结束时间
            byte[] startRow = Bytes.toBytes(eqDto.getEquipNo() + startTime);
            byte[] endRow = Bytes.toBytes(eqDto.getEquipNo() + endTime);
            scan.withStartRow(startRow, true);
            scan.withStopRow(endRow, true);

            //查询列
            scan.addColumn(columnFamily, Bytes.toBytes(columnName));
            logger.info("scan columnFamily[{}],startRow[{}],endRow[{}]",Bytes.toString(columnFamily),Bytes.toString(startRow),Bytes.toString(endRow));

            //翻转查询
            scan.setReversed(true);
            scan.setCacheBlocks(false);
            scan.setCaching(2000);

            List<EquipParamDataDTO> listDatas = hbaseTemplate.find(TABLE_NAME, scan, (result, rowNum) -> {
                EquipParamDataDTO eqParamData = new EquipParamDataDTO();
                ParamDataDto paramData = new ParamDataDto();
                String rowKey = Arrays.toString(result.getRow());
                eqParamData.setRowKey(rowKey);
                List<Cell> cells = result.listCells();
                //过滤列
                for (Cell cell : cells) {
                    String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                    if (columnName.equals(key)) {
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        paramData.setParamCode(param.getParamId());
                        paramData.setValue(value);
                        eqParamData.setParamData(paramData);
                        // 填充数值
                        param.setValue(value);
                    }
                }
                return eqParamData;
            });

            if (listDatas == null||listDatas.size()==0) {
                param.setValue("");
            }


        }
        //TODO 目前还未确认上线时间
        String equipCode = eqDto.getEquipNo();
        if (StringUtils.isBlank(equipCode)) {
            eqDto.setStatus("0");
        } else {
            String status = ttEqConfigurationDao.getEquipStatus(equipCode);
            if (StringUtils.isNotBlank(status)) {
                eqDto.setStatus(status);
            } else {
                // 默认状态
                eqDto.setStatus("0");
            }
        }
        return eqDto;
    }

    /**
     * 查询列簇名
     * @param equipCode
     * @param pkId
     * @return
     */
    private byte[] getColumnFamily(String equipCode, String pkId) {
        if (StringUtils.isBlank(equipCode) || StringUtils.isBlank(pkId)) {
            return null;
        }
        Map<String, Object> param = new HashMap<>();
        param.put("equipCode", equipCode);
        param.put("pkId", pkId);
        List<Map<String, Object>> list = this.ttEqConfigurationDao.getHBaseQueryParam(param);
        if (list != null && list.size() > 0) {
            return Bytes.toBytes(list.get(0).get("rowName").toString());
        } else {
            return null;
        }
    }

    /**
     * 查询列明
     * @param equipCode
     * @param pkId
     * @return
     */
    private String getColumnName(String equipCode, String pkId) {
        if (StringUtils.isBlank(equipCode) || StringUtils.isBlank(pkId)) {
            return null;
        }
        Map<String, Object> param = new HashMap<>();
        param.put("equipCode", equipCode);
        param.put("pkId", pkId);
        List<Map<String, Object>> list = this.ttEqConfigurationDao.getHBaseQueryParam(param);
        if (list != null && list.size() > 0 && list.get(0).get("param")!=null) {
            return list.get(0).get("param").toString();
        } else {
            return null;
        }
    }

    @Override
    public void insertData(final String id) {
        assert StringUtils.isNotBlank(id);
        hbaseTemplate.execute("MachineData", htable -> {
            List<ConfigurationEqDto> list = getConfigurationEqLists(id);
            List<Put> puts = new ArrayList<>();
            list.forEach(eqDto -> {
                byte[] key = Bytes.toBytes(eqDto.getEquipNo() + Calendar.getInstance().getTime().getTime());
                List<ConfigurationParamDto> params = eqDto.getProperties();
                Put put = new Put(key);
                params.forEach(param -> put.addColumn(getColumnFamily(eqDto.getEquipNo(), param.getParamId()), Bytes.toBytes(getColumnName(eqDto.getEquipNo(), param.getParamId())), Bytes.toBytes(String.valueOf(getValue()))));
                puts.add(put);
            });
            htable.put(puts);
            return true;
        });
    }

    public static void main(String[] args) {
        Date date = new Date();
        date.setTime(1536669180025L);
        SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println(f.format(date));
    }
}
